package com.gridmobi.xrtb.handler.impl;


import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.gridmobi.xrtb.common.ExecutorPool;
import com.gridmobi.xrtb.domain.dsp.DspTrackClickLog;
import com.gridmobi.xrtb.domain.dsp.DspTrackImpLog;
import com.gridmobi.xrtb.domain.dsp.DspTrackLog;
import com.gridmobi.xrtb.domain.dsp.DspTrackWinLog;
import com.gridmobi.xrtb.domain.dsp.common.AppConstant;
import com.gridmobi.xrtb.domain.dsp.common.KafkaConstant;
import com.gridmobi.xrtb.domain.dsp.common.enums.BidTrackTypeE;
import com.gridmobi.xrtb.domain.dsp.common.enums.CreativeTypeE;
import com.gridmobi.xrtb.handler.BiddingLogger;
import com.gridmobi.xrtb.model.openrtb23.*;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;

import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

//@Component("kafkaBiddingLogger")
public class KafkaBiddingLogger implements BiddingLogger {
    private static final Logger errorlogger = LoggerFactory.getLogger("error");
    private static final Logger senderror = LoggerFactory.getLogger("senderror");

    @Autowired
    private BidHandler bidHandler;
    @Autowired
    private ObjectMapper jsonMapper;
    @Autowired
    @Qualifier("rtbproducer")
    private Producer producer;


    @Autowired
    private SysError sysError;


    AtomicInteger count = new AtomicInteger();
    Random random = new Random();

    @Override
    public void requestlog(String os ,String deviceid, String ip, String appid) {
        ExecutorPool.getExecutor().execute(() -> {
            try {
                producer.send(new ProducerRecord(KafkaConstant.RTB_REQUESTLOG_TOPIC,
                                os
                                .concat(AppConstant.SEPARATOR)
                                .concat(appid)
                                .concat(AppConstant.SEPARATOR)
                                .concat(ip)
                                .concat(AppConstant.SEPARATOR)
                                .concat(deviceid)));
            } catch (Exception e) {
                //如果报错 打印log 增加定时任务重发log
                errorlogger.error("",e);

            }
        });
    }

    @Override
    public void bidlogSend(DspTrackLog dspTrackLog) {
        ExecutorPool.getExecutor().execute(() -> {
            try {
                String log = jsonMapper.writeValueAsString(dspTrackLog);
                producer.send(new ProducerRecord(KafkaConstant.RTB_BIDLOG_TOPIC, String.valueOf(random.nextInt(10)), log));
            } catch (Exception e) {
                //如果报错 打印log 增加定时任务重发log
                senderror.info(BidTrackTypeE.BID.v + JSONObject.toJSONString(dspTrackLog));
                sysError.kafkaError(e);

            }
        });

    }

    @Override
    public void bidlog(OpenRTB23BidResponse resp, OpenRTB23BidRequest bidRequest) {
        DspTrackLog dspTrackLog = bidRequest.getDspTrackLog();
        for (SeatBid seatBid : resp.getSeatbid()) {
            for (Bid bid : seatBid.getBid()) {
                dspTrackLog.setBidId(bid.getId());
                dspTrackLog.setCid(bid.getCid());
                dspTrackLog.setCrid(bid.getCrid());
                dspTrackLog.setPrice(String.valueOf(bid.getPrice()));
                dspTrackLog.setImpid(bid.getImpid());


                Imp imp = bidHandler.getBidImp(bid.getImpid(), bidRequest);
                char c = bid.getCrid().charAt(0);
                /*
                 * 获取 imp pos 广告位置
                 */
                if (CreativeTypeE.BANNER.code == c || CreativeTypeE.HTML.code == c) {
                    dspTrackLog.setPos(String.valueOf(imp.getBanner().getPos()));
                }
                if (CreativeTypeE.VIDEO.code == c || CreativeTypeE.REWARDEDVIDEO.code == c) {
                    dspTrackLog.setPos(String.valueOf(imp.getVideo().getPos()));

                }
                dspTrackLog.setW(bid.getW());
                dspTrackLog.setH(bid.getH());
                bidlogSend(dspTrackLog);


            }
        }


    }

    @Override
    public void winlog(DspTrackLog dspTrackLog) {
        try {
            DspTrackWinLog dspTrackWinLog = new DspTrackWinLog();
            BeanUtils.copyProperties(dspTrackLog, dspTrackWinLog);
            if (StringUtils.isBlank(dspTrackWinLog.getBidstmp())) {
                dspTrackWinLog.setBidstmp(String.valueOf(System.currentTimeMillis()));
            }
            String log = jsonMapper.writeValueAsString(dspTrackWinLog);

            producer.send(new ProducerRecord(KafkaConstant.RTB_WINLOG_TOPIC, String.valueOf(random.nextInt(10)), log));

            SysError.KAFKA_ERROR_TIMES.set(0);

        } catch (Exception e) {
            //离线打印 发送失败的数据 ，用任务读文件去补发
            senderror.info(BidTrackTypeE.WIN.v + JSONObject.toJSONString(dspTrackLog));
            sysError.kafkaError(e);

            errorlogger.error(JSONObject.toJSONString(dspTrackLog), e);
        }
    }


    @Override
    public void losslog(DspTrackLog dspTrackLog) {
        try {
            DspTrackWinLog dspTrackWinLog = new DspTrackWinLog();
            BeanUtils.copyProperties(dspTrackLog, dspTrackWinLog);
            producer.send(new ProducerRecord(KafkaConstant.RTB_LOSSLOG_TOPIC, String.valueOf(random.nextInt(10)), jsonMapper.writeValueAsString(dspTrackWinLog)));
        } catch (JsonProcessingException e) {
            sysError.kafkaError(e);

            errorlogger.error(JSONObject.toJSONString(dspTrackLog), e);
        }
    }


    @Override
    public void imprlog(DspTrackLog dspTrackLog) {
        try {
            DspTrackImpLog dspTrackImpLog = new DspTrackImpLog();
            BeanUtils.copyProperties(dspTrackLog, dspTrackImpLog);
            if (StringUtils.isBlank(dspTrackImpLog.getBidstmp())) {
                dspTrackImpLog.setBidstmp(String.valueOf(System.currentTimeMillis()));
            }
            String log = jsonMapper.writeValueAsString(dspTrackImpLog);
            producer.send(new ProducerRecord(KafkaConstant.RTB_IMPRLOG_TOPIC, String.valueOf(random.nextInt(10)), log));

            SysError.KAFKA_ERROR_TIMES.set(0);

        } catch (Exception e) {
            //离线打印 发送失败的数据 ，用任务读文件去补发
            senderror.info(BidTrackTypeE.IMP.v + JSONObject.toJSONString(dspTrackLog));
            sysError.kafkaError(e);
            errorlogger.error(JSONObject.toJSONString(dspTrackLog), e);
        }
    }

    @Override
    public void clicklog(DspTrackLog dspTrackLog) {

        ExecutorPool.getExecutor().execute(() -> {
            try {
                DspTrackClickLog dspTrackClickLog = new DspTrackClickLog();
                BeanUtils.copyProperties(dspTrackLog, dspTrackClickLog);
                if (StringUtils.isBlank(dspTrackClickLog.getBidstmp())) {
                    dspTrackClickLog.setBidstmp(String.valueOf(System.currentTimeMillis()));
                }
                String log = jsonMapper.writeValueAsString(dspTrackClickLog);
                producer.send(new ProducerRecord(KafkaConstant.RTB_CLICKLOG_TOPIC, String.valueOf(random.nextInt(10)), log));
                SysError.KAFKA_ERROR_TIMES.set(0);

            } catch (Exception e) {
                //离线打印 发送失败的数据 ，用任务读文件去补发
                senderror.info(BidTrackTypeE.CLICK.v + JSONObject.toJSONString(dspTrackLog));
                sysError.kafkaError(e);
                errorlogger.error(JSONObject.toJSONString(dspTrackLog), e);
            }
        });

    }

    @Override
    public void landlog(DspTrackLog dspTrackLog) {
        try {
            DspTrackClickLog dspTrackClickLog = new DspTrackClickLog();
            BeanUtils.copyProperties(dspTrackLog, dspTrackClickLog);
            producer.send(new ProducerRecord(KafkaConstant.RTB_LANDLOG_TOPIC, String.valueOf(random.nextInt(10)), jsonMapper.writeValueAsString(dspTrackClickLog)));
        } catch (Exception e) {

            sysError.kafkaError(e);

            errorlogger.error(JSONObject.toJSONString(dspTrackLog), e);
        }
    }
}
